package com.newland.nxsms;

import com.newland.nxsms.bean.CmppFromMsg;
import com.newland.nxsms.bean.CmppReport;
import com.newland.nxsms.cmpp.CMPPCommon;
import com.newland.nxsms.cmpp.CMPPDao;
import com.newland.nxsms.cmpp.CMPPReport;
import com.newland.nxsms.dao.ConnectionManager;
import com.newland.nxsms.utils.DbUtils;
import com.newland.nxsms.utils.RedisCluster;
import com.newland.nxsms.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.Connection;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author zhangheng
 * @date 2020-10-27 19:28:25
 */
public class RespMsgToDb implements Runnable {
	private static final Logger logger = LoggerFactory.getLogger(RespMsgToDb.class);
	private CMPPDao dao;
	public RespMsgToDb(CMPPDao dao) {
		this.dao = dao;
	}
	private RedisCluster cluster;
	private AtomicLong atomicLong = new AtomicLong(0L);

	public AtomicLong getAtomicLong() {
		return atomicLong;
	}
	public void setAtomicLong(Long i) {
		this.atomicLong.set(i); ;
	}
	@Override
	public void run() {
		try {
			cluster = new RedisCluster(NxSmsMain.REDIS_CFG);
		} catch (IOException e) {
			logger.error("respMsg redis init failed :"+e.getMessage());
			e.printStackTrace();
		}
		while(true) {
			Connection conn = null;
			try {
				conn= ConnectionManager.getInstance().getConnection();
				dealReport(conn);
				dealFromMsg(conn);
			} catch (Exception e) {
				logger.error("respMsgToDb error :"+e.getMessage());
				e.printStackTrace();
			}finally {
				DbUtils.closeQuietly(conn);
			}
		}
	}
	private void dealReport(Connection conn) throws Exception {
		try{
			CmppReport cmppReport;
			cmppReport = NxSmsMain.reports.poll();
			if (null != cmppReport) {
				this.getAtomicLong().incrementAndGet();
				CMPPReport report = cmppReport.getReport();
				dao.insertCMPPReport(conn, false, cmppReport.getSeq(), report);
				String msgId = CMPPCommon.msgIdToSting(report.getMsgId())+"";
				String seqId = cluster.get(NxSmsMain.HOST_NAME+"MsgId+" + msgId);
				if (StringUtils.isNotBlank(seqId)){
					cluster.expire(NxSmsMain.HOST_NAME+"MsgId+" + msgId,0);
					String parentSeqId = cluster.get(NxSmsMain.HOST_NAME+"sendLongMsg" + seqId);
					if (StringUtils.isBlank(parentSeqId)){
						parentSeqId = seqId;
					}
					cluster.expire(NxSmsMain.HOST_NAME+"sendLongMsg" + seqId,0);
					if (!"DELIVRD".equals(report.getStat())) {
						dao.updateSmsReportDetail(conn, false, Integer.parseInt(parentSeqId));
						cluster.setex(NxSmsMain.HOST_NAME+"updateReport"+parentSeqId,"-1",2*24*60*60);
					}
				}
			} else {
				Thread.sleep(100);
			}
		} catch (Exception e) {
			throw  e;
		}
	}
	private void dealFromMsg(Connection conn) throws Exception {
		try {
			CmppFromMsg fromMsg;
			fromMsg = NxSmsMain.fromMsgs.poll();
			if (null!=fromMsg){
				dao.insertCMPPDeliver(conn, false, fromMsg.getSeq(), fromMsg.getDeliver());
			}else {
				Thread.sleep(100);
			}
		} catch (Exception e) {
			throw  e;
		}
	}

}
